/**
 * @author zhounan 
 * @date 2024年6月19日
 * @version V1.0
 */
package org.zn.etl.job.context;

import java.util.List;

import javax.annotation.PostConstruct;

import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.BoundHashOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.zn.etl.misc.BeanCopierUtils;

import lombok.extern.log4j.Log4j2;

/**
 * @author zhounan 
 * @date 2024年6月19日
 * @version V1.0
 */
@Log4j2
public class RedisClusterContextManager  implements ClusterContextManager {
	
	@Autowired
	private RedisTemplate<String, Object> redisTemplate;
	
	private BoundHashOperations<String, String, ClusterJobExecInfo>  hashOps;
	
	private static final String ETL_JOBS_KEY="ETL_JOBRUNNINGMANAGER";
	
	//private ObjectMapper objectMapper;
	
	@PostConstruct
	public void init() {
		this.hashOps=redisTemplate.boundHashOps(ETL_JOBS_KEY);
		
		log.info("RedisClusterContextManager init");
		
	}

	public void addJobExecIns(JobExecInstanceContext ins) throws Exception {
		
		ClusterJobExecInfo ni=BeanCopierUtils.copy(ins, new ClusterJobExecInfo());
		
		ni.setTotal((Long) ins.getContextMap().get(GlobalContextKey.TOTALROWS));
		
		if(StringUtils.isNotEmpty(ins.getClusterJobName())) {
			ni.setJobName(ins.getClusterJobName());
		}else {
			ni.setJobName(ins.getJob().getJobName());			
		}
		ni.setExecParam(null);

		boolean  put=hashOps.putIfAbsent(ni.getJobName(), ni);
		
		if(!put) {	
			throw new Exception("job:"+ni.getJobName()+" has running in cluster");
		}
		
	}
	
	public void updateJobExecIns(JobExecInstanceContext ins)   {
		
		ClusterJobExecInfo ni=BeanCopierUtils.copy(ins, new ClusterJobExecInfo());
		ni.setTotal((Long) ins.getContextMap().get(GlobalContextKey.TOTALROWS));
		
	
		if(StringUtils.isNotEmpty(ins.getClusterJobName())) {
			ni.setJobName(ins.getClusterJobName());
		}else {
			ni.setJobName(ins.getJob().getJobName());			
		}
		
		//ni.setExecParam(null);

		if(hashOps.hasKey(ni.getJobName())) {
			redisTemplate.boundHashOps(ETL_JOBS_KEY).put(ni.getJobName(), ni);			
		}
		
		
		
	}
	
	
	public List<ClusterJobExecInfo> getAll() {
		List<ClusterJobExecInfo> jobList=hashOps.values();
		return jobList;
	}

	public ClusterJobExecInfo find(String jobName) {
		return hashOps.get(jobName);
	}

	@Override
	public boolean remove(String jobName) {
		// TODO Auto-generated method stub
		
		return hashOps.delete(jobName)>0;
	}
	

}
